package com.xinpark.shared.message;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.javaapi.producer.ProducerData;
import kafka.producer.ProducerConfig;

public class MsgProducer {

    private Producer<Integer, String> producer = null;
    private Properties                props;

    public void init(){
        // Use random partitioner. Don't need the key type. Just set it to Integer.
        // The message is of type String.
        producer = new Producer<Integer, String>(new ProducerConfig(props));
    }
    
    public void send(final List<String> topicList, final byte [] message){
        try{
            List<ProducerData<Integer, String>> dataForMultipleTopics = new ArrayList<ProducerData<Integer, String>>();
            for(String topic : topicList){
                ProducerData<Integer, String> data = new ProducerData<Integer, String>(topic, new String(message));
                dataForMultipleTopics.add(data);
            }

            producer.send(dataForMultipleTopics);
        }catch(Exception e){
            System.out.println("Producer reconnect.......");
            init();
        }
        return;
    }
    
    public void setProps(Properties props) {
        this.props = props;
    }

}
